package drds.binlog.kafka;

import drds.binlog.common.position.ClientInfo;
import drds.binlog.kafka.KafkaProperties.Destination;
import drds.binlog.kafka.KafkaProperties.Topic;
import drds.binlog.protocol.Message;
import drds.binlog.server.ServerStarter;
import drds.binlog.server.embedded.EmbeddedServer;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.yaml.snakeyaml.Yaml;

import java.io.FileInputStream;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * kafka 启动类
 */
public class KafkaStarter implements ServerStarter
{

    private static final Logger logger = LoggerFactory.getLogger(KafkaStarter.class);

    private static final String CLASSPATH_URL_PREFIX = "classpath:";

    private volatile boolean running = false;

    private ExecutorService executorService;

    private KafkaProducer kafkaProducer;

    private KafkaProperties kafkaProperties;

    public void init()
    {
        try
        {
            logger.info("## load kafka configurations");
            String conf = System.getProperty("kafka.conf", "classpath:kafka.yml");

            if (conf.startsWith(CLASSPATH_URL_PREFIX))
            {
                conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX);
                kafkaProperties = new Yaml().loadAs(KafkaStarter.class.getClassLoader().getResourceAsStream(conf),
                        KafkaProperties.class);
            } else
            {
                kafkaProperties = new Yaml().loadAs(new FileInputStream(conf), KafkaProperties.class);
            }

            // 初始化 kafka producer
            kafkaProducer = new KafkaProducer();
            kafkaProducer.init(kafkaProperties);
            // set filterTransactionEntry
            // if (kafkaProperties.isFilterTransactionEntry()) {
            // System.setProperty("canal.instance.eventFilter.transaction.entry", "true");
            // }
            // 对应每个instance启动一个worker线程
            List<KafkaProperties.Destination> destinationList = kafkaProperties.getDestinations();

            executorService = Executors.newFixedThreadPool(destinationList.size());

            logger.info("## start the kafka workers.");
            for (final Destination destination : destinationList)
            {
                executorService.execute(new Runnable()
                {

                    @Override
                    public void run()
                    {
                        worker(destination);
                    }
                });
            }
            running = true;
            logger.info("## the kafka workers is running now ......");
            Runtime.getRuntime().addShutdownHook(new Thread()
            {

                public void run()
                {
                    try
                    {
                        logger.info("## stop the kafka workers");
                        running = false;
                        executorService.shutdown();
                        kafkaProducer.stop();
                    } catch (Throwable e)
                    {
                        logger.warn("##something goes wrong when stopping kafka workers:", e);
                    } finally
                    {
                        logger.info("## canal kafka is down.");
                    }
                }

            });

        } catch (Throwable e)
        {
            logger.error("## Something goes wrong when starting up the canal kafka workers:", e);
            System.exit(0);
        }
    }

    private void worker(Destination destination)
    {
        while (!running)
            ;
        logger.info("## start the canal consumer: {}.", destination.getCanalDestination());
        final EmbeddedServer embeddedServer = EmbeddedServer.instance();
        final ClientInfo clientInfo = new ClientInfo(destination.getCanalDestination(), (short) 1001, "");
        while (running)
        {
            try
            {
                if (!embeddedServer.getDestinationToInstanceMap().containsKey(clientInfo.getDestination()))
                {
                    try
                    {
                        Thread.sleep(3000);
                    } catch (InterruptedException e)
                    {
                        // ignore
                    }
                    continue;
                }
                embeddedServer.subscribe(clientInfo);
                logger.info("## the canal consumer {} is running now ......", destination.getCanalDestination());

                while (running)
                {
                    Message message;
                    if (kafkaProperties.getCanalGetTimeout() != null)
                    {
                        message = embeddedServer.getWithoutAck(clientInfo,
                                kafkaProperties.getCanalBatchSize(),
                                kafkaProperties.getCanalGetTimeout(),
                                TimeUnit.MILLISECONDS);
                    } else
                    {
                        message = embeddedServer.getWithoutAck(clientInfo, kafkaProperties.getCanalBatchSize());
                    }

                    final long batchId = message.getId();
                    try
                    {
                        int size = message.isRaw() ? message.getRawEntryList().size() : message.getEntryList().size();
                        if (batchId != -1 && size != 0)
                        {
                            Topic topic = new Topic();
                            topic.setTopic(destination.getTopic());
                            topic.setPartition(destination.getPartition());
                            kafkaProducer.send(topic, message, new KafkaProducer.Callback()
                            {

                                @Override
                                public void commit()
                                {
                                    embeddedServer.ack(clientInfo, batchId);
                                }

                                @Override
                                public void rollback()
                                {
                                    embeddedServer.rollback(clientInfo, batchId);
                                }
                            });
                        } else
                        {
                            try
                            {
                                Thread.sleep(100);
                            } catch (InterruptedException e)
                            {
                                // ignore
                            }
                        }

                    } catch (Exception e)
                    {
                        logger.error(e.getMessage(), e);
                    }
                }
            } catch (Exception e)
            {
                logger.error("process error!", e);
            }
        }
    }
}
